package com.wis.bolt.redis;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.wis.TopologyDrive;
import com.wis.pool.redisCluster.MyJedisClusterPool;
import com.wis.tools.CommonUtiles;
import com.wis.tools.MPGUtiles;
import com.wis.traffic.common.string.StringUtil;
import com.wis.traffic.entity.dao.DataTimes;
import com.wis.traffic.post.service.DataTimesService;
import com.wis.traffic.post.service.impl.DataTimesServiceImpl;
import org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import redis.clients.jedis.JedisCluster;

import java.util.Map;

/**
 * @Description: traffic_wis
 * @author: fan
 * @Date: Created in 2019/8/4 9:33
 * @Modified By:
 */
public class StoreData extends BaseRichBolt {
    private OutputCollector collector;
    private static String typeProduce;
    private static Map map;
    private static JedisCluster redisClusterObject;
    private static DataTimesService dataTimesService;

    @Override
    public void prepare(Map map, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.map = map;
        dataTimesService = new DataTimesServiceImpl();
        typeProduce = map.get(TopologyDrive.TYPE_PRODUCE).toString();
        try {
            if (redisClusterObject == null) redisClusterObject = MyJedisClusterPool.getRedisClusterObject();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            MyJedisClusterPool.returnJedisCluterObject(redisClusterObject);
        }
    }

    @Override
    public void execute(Tuple tuple) {
        String key = tuple.getStringByField("key");
        String value = tuple.getStringByField("value");
        if (redisClusterObject == null) MyJedisClusterPool.getRedisClusterObject();
        if (value.contains("over")) {
            DataTimes dataTimes = new DataTimes();
            dataTimes.setData_type(Integer.parseInt(value.substring(0, 1)));
            dataTimes.setGeo_type(Integer.parseInt(value.substring(2, 3)));
            dataTimes.setProduct_type(Integer.parseInt(value.substring(4, 5)));
            dataTimes.setUpdated_time(key.substring(9, 21));
            dataTimesService.insertOrUpdate(dataTimes);
            MyJedisClusterPool.returnJedisCluterObject(redisClusterObject);
            System.out.println("----------------");
        } else if (StringUtils.isNotBlank(value)) {
            String first = key.substring(0, key.length() - 2);
            String second = key.substring(key.length() - 2);
            if (redisClusterObject.exists(first)) {
                value = CommonUtiles.mergeValue(first, second, value, redisClusterObject);
            }
            if (StringUtils.isNotBlank(value)) {
                redisClusterObject.hset(first, second, value);
                redisClusterObject.expire(first, 21600);
            }
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        MyJedisClusterPool.returnJedisCluterObject(redisClusterObject);
    }
}
